/*-------------------------------------------------------------------------
 *
 * pqmq.c
 *	  使用前端/后端协议通过 shm_mq 进行通信
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *	src/backend/libpq/pqmq.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"

static shm_mq_handle *pq_mq_handle;
static bool pq_mq_busy = false;
static pid_t pq_mq_parallel_leader_pid = 0;
static pid_t pq_mq_parallel_leader_backend_id = InvalidBackendId;

static void fc_pq_cleanup_redirect_to_shm_mq(dsm_segment *fc_seg, Datum fc_arg);
static void fc_mq_comm_reset(void);
static int	fc_mq_flush(void);
static int	fc_mq_flush_if_writable(void);
static bool fc_mq_is_send_pending(void);
static int	fc_mq_putmessage(char fc_msgtype, const char *fc_s, size_t fc_len);
static void fc_mq_putmessage_noblock(char fc_msgtype, const char *fc_s, size_t fc_len);

static const PQcommMethods PqCommMqMethods = {
	fc_mq_comm_reset,
	fc_mq_flush,
	fc_mq_flush_if_writable,
	fc_mq_is_send_pending,
	fc_mq_putmessage,
	fc_mq_putmessage_noblock
};

/*
 * 安排将前端/后端协议消息重定向到共享内存
 * 消息队列。
 */
void pq_redirect_to_shm_mq(dsm_segment *fc_seg, shm_mq_handle *fc_mqh)
{
	PqCommMethods = &PqCommMqMethods;
	pq_mq_handle = fc_mqh;
	whereToSendOutput = DestRemote;
	FrontendProtocol = PG_PROTOCOL_LATEST;
	on_dsm_detach(fc_seg, fc_pq_cleanup_redirect_to_shm_mq, (Datum) 0);
}

/*
 * 当包含我们的 shm_mq 的 DSM 消失时，我们需要停止发送
 * 消息到它。
 */
static void fc_pq_cleanup_redirect_to_shm_mq(dsm_segment *fc_seg, Datum fc_arg)
{
	pq_mq_handle = NULL;
	whereToSendOutput = DestNone;
}

/*
 * 安排每次通过 shm_mq 传输消息数据时向并行领导发送
 * SendProcSignal()。
 */
void pq_set_parallel_leader(pid_t fc_pid, BackendId fc_backend_id)
{
	Assert(PqCommMethods == &PqCommMqMethods);
	pq_mq_parallel_leader_pid = fc_pid;
	pq_mq_parallel_leader_backend_id = fc_backend_id;
}

static void fc_mq_comm_reset(void)
{
	/* 没有什么可做的。 */
}

static int fc_mq_flush(void)
{
	/* 没有什么可做的。 */
	return 0;
}

static int fc_mq_flush_if_writable(void)
{
	/* 没有什么可做的。 */
	return 0;
}

static bool fc_mq_is_send_pending(void)
{
	/* 从来没有任何待处理的事项。 */
	return 0;
}

/*
 * 将 libpq 协议消息传输到通过 pq_mq_handle 选择的共享内存消息队列。
 * 我们不包括长度字，因为接收方将通过 shm_mq_receive() 知道消息的长度。
 */
static int fc_mq_putmessage(char fc_msgtype, const char *fc_s, size_t fc_len)
{
	shm_mq_iovec fc_iov[2];
	shm_mq_result fc_result;

	/*
	 * 如果我们正在发送消息，并且由于队列已满而不得不等待，
	 * 然后我们被中断，而该中断导致尝试发送另一条消息，
	 * 我们的反应是分离队列。没有办法返回到原始上下文，
	 * 但即使有，只是将消息排队也将无限期推迟对中断的响应。
	 * 所以我们选择这样做。
	 */
	if (pq_mq_busy)
	{
		if (pq_mq_handle != NULL)
			shm_mq_detach(pq_mq_handle);
		pq_mq_handle = NULL;
		return EOF;
	}

	/*
	 * 如果消息队列已经消失，就忽略该消息。这
	 * 不一定表示有问题；例如，DEBUG 消息可能
	 * 在关闭序列的后期生成，在所有 DSM 已经
	 * 被分离之后。
	 */
	if (pq_mq_handle == NULL)
		return 0;

	pq_mq_busy = true;

	fc_iov[0].data = &fc_msgtype;
	fc_iov[0].len = 1;
	fc_iov[1].data = fc_s;
	fc_iov[1].len = fc_len;

	Assert(pq_mq_handle != NULL);

	for (;;)
	{
		/*
		 * 通过将 force_flush 传递为 true 立即通知接收方，
		 * 以便在我们发送并行消息信号之前更新共享内存值。
		 */
		fc_result = shm_mq_sendv(pq_mq_handle, fc_iov, 2, true, true);

		if (pq_mq_parallel_leader_pid != 0)
			SendProcSignal(pq_mq_parallel_leader_pid,
						   PROCSIG_PARALLEL_MESSAGE,
						   pq_mq_parallel_leader_backend_id);

		if (fc_result != SHM_MQ_WOULD_BLOCK)
			break;

		(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
						 WAIT_EVENT_MQ_PUT_MESSAGE);
		ResetLatch(MyLatch);
		CHECK_FOR_INTERRUPTS();
	}

	pq_mq_busy = false;

	Assert(fc_result == SHM_MQ_SUCCESS || fc_result == SHM_MQ_DETACHED);
	if (fc_result != SHM_MQ_SUCCESS)
		return EOF;
	return 0;
}

static void fc_mq_putmessage_noblock(char fc_msgtype, const char *fc_s, size_t fc_len)
{
	/*
	 * 虽然 shm_mq 机制确实支持非阻塞模式发送消息，
	 * 但目前没有办法尝试发送开始发送消息而不
	 * 使我们承诺完成传输。这在未来可以改善，
	 * 但现在我们不需要它。
	 */
	elog(ERROR, "not currently supported");
}

/*
 * 解析 ErrorResponse 或 NoticeResponse 负载，并用结果填充
 * ErrorData 结构体。
 */
void pq_parse_errornotice(StringInfo fc_msg, ErrorData *fc_edata)
{
	/* 用合理的默认值初始化 edata。 */
	MemSet(fc_edata, 0, sizeof(ErrorData));
	fc_edata->elevel = ERROR;
	fc_edata->assoc_context = CurrentMemoryContext;

	/* 循环遍历字段并提取每一个。 */
	for (;;)
	{
		char		fc_code = pq_getmsgbyte(fc_msg);
		const char *fc_value;

		if (fc_code == '\0')
		{
			pq_getmsgend(fc_msg);
			break;
		}
		fc_value = pq_getmsgrawstring(fc_msg);

		switch (fc_code)
		{
			case PG_DIAG_SEVERITY:
				/* 忽略，信任我们将获得非本地化版本 */
				break;
			case PG_DIAG_SEVERITY_NONLOCALIZED:
				if (strcmp(fc_value, "DEBUG") == 0)
				{
					/*
					 * 我们无法重建确切的 DEBUG 级别，但
					 * 假设它是 >= client_min_messages，因此选择
					 * DEBUG1 以确保我们会将其传递给客户端。
					 */
					fc_edata->elevel = DEBUG1;
				}
				else if (strcmp(fc_value, "LOG") == 0)
				{
					/*
					 * 它不能是 LOG_SERVER_ONLY，否则工作者不会
					 * 将其发送给我们；因此 LOG 是正确的值。
					 */
					fc_edata->elevel = LOG;
				}
				else if (strcmp(fc_value, "INFO") == 0)
					fc_edata->elevel = INFO;
				else if (strcmp(fc_value, "NOTICE") == 0)
					fc_edata->elevel = NOTICE;
				else if (strcmp(fc_value, "WARNING") == 0)
					fc_edata->elevel = WARNING;
				else if (strcmp(fc_value, "ERROR") == 0)
					fc_edata->elevel = ERROR;
				else if (strcmp(fc_value, "FATAL") == 0)
					fc_edata->elevel = FATAL;
				else if (strcmp(fc_value, "PANIC") == 0)
					fc_edata->elevel = PANIC;
				else
					elog(ERROR, "unrecognized error severity: \"%s\"", fc_value);
				break;
			case PG_DIAG_SQLSTATE:
				if (strlen(fc_value) != 5)
					elog(ERROR, "invalid SQLSTATE: \"%s\"", fc_value);
				fc_edata->sqlerrcode = MAKE_SQLSTATE(fc_value[0], fc_value[1], fc_value[2],
												  fc_value[3], fc_value[4]);
				break;
			case PG_DIAG_MESSAGE_PRIMARY:
				fc_edata->message = pstrdup(fc_value);
				break;
			case PG_DIAG_MESSAGE_DETAIL:
				fc_edata->detail = pstrdup(fc_value);
				break;
			case PG_DIAG_MESSAGE_HINT:
				fc_edata->hint = pstrdup(fc_value);
				break;
			case PG_DIAG_STATEMENT_POSITION:
				fc_edata->cursorpos = pg_strtoint32(fc_value);
				break;
			case PG_DIAG_INTERNAL_POSITION:
				fc_edata->internalpos = pg_strtoint32(fc_value);
				break;
			case PG_DIAG_INTERNAL_QUERY:
				fc_edata->internalquery = pstrdup(fc_value);
				break;
			case PG_DIAG_CONTEXT:
				fc_edata->context = pstrdup(fc_value);
				break;
			case PG_DIAG_SCHEMA_NAME:
				fc_edata->schema_name = pstrdup(fc_value);
				break;
			case PG_DIAG_TABLE_NAME:
				fc_edata->table_name = pstrdup(fc_value);
				break;
			case PG_DIAG_COLUMN_NAME:
				fc_edata->column_name = pstrdup(fc_value);
				break;
			case PG_DIAG_DATATYPE_NAME:
				fc_edata->datatype_name = pstrdup(fc_value);
				break;
			case PG_DIAG_CONSTRAINT_NAME:
				fc_edata->constraint_name = pstrdup(fc_value);
				break;
			case PG_DIAG_SOURCE_FILE:
				fc_edata->filename = pstrdup(fc_value);
				break;
			case PG_DIAG_SOURCE_LINE:
				fc_edata->lineno = pg_strtoint32(fc_value);
				break;
			case PG_DIAG_SOURCE_FUNCTION:
				fc_edata->funcname = pstrdup(fc_value);
				break;
			default:
				elog(ERROR, "unrecognized error field code: %d", (int) fc_code);
				break;
		}
	}
}
